package com.foreveross.project.projectonline.jfinal.plugin;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.foreveross.project.projectonline.jfinal.plugin.JedisSentinelKit.JedisMessage;
import com.google.common.collect.Maps;

public class JedisSentinelQueueConsumer {

	static final String SEPARATER = "-";
	static final String PROCESSINGLIST = SEPARATER + "processing";
	static final String PREFIX = "queue" + SEPARATER;

	static Map<String, JedisSentinelQueueConsumer> existingConsumers = Maps
			.newHashMap();

	int interval = 1000;

	boolean start = false;

	String queueName;

	private JedisSentinelQueueConsumer(String queueName) {
		this.queueName = queueName;
	}

	public static JedisSentinelQueueConsumer create(String queueName) {
		JedisSentinelQueueConsumer consumer = existingConsumers.get(queueName);
		if (consumer == null) {
			consumer = new JedisSentinelQueueConsumer(queueName);
		} else {
			throw new IllegalArgumentException("The consumer named "
					+ queueName + " already exists");
		}
		return consumer;
	}

	public JedisSentinelQueueConsumer interval(int interval) {
		this.interval = interval;
		return this;
	}

	private void waitForMessages() {
		try {
			TimeUnit.MILLISECONDS.sleep(interval);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public <T> T consume() {
		return JedisSentinelKit.rpoplpush(queueName(), processingListName());
	}

	@SuppressWarnings( { "unchecked" })
	public void consume(final JedisMessage callback) {
		if (start) {
			throw new RuntimeException("The Consumer named " + queueName
					+ " is working");
		}
		new Thread(new Runnable() {
			public void run() {
				while (true) {
					JedisSentinelKit.rpoplpush(queueName(),
							processingListName());
					List message = JedisSentinelKit.lrange(
							processingListName(), -1, -1);
					if (message.isEmpty()) {
						waitForMessages();
					} else {
						callback.onMessage(message.get(0));
						JedisSentinelKit.rpop(processingListName());
					}
				}
			}
		}).start();
	}

	private String processingListName() {
		return PREFIX + queueName + PROCESSINGLIST;
	}

	private String queueName() {
		return PREFIX + queueName;
	}

}
